package org.dragonnova.iot.business.cloud.mq.service;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SettableListenableFuture;

/**
 *
 * @description: Kafk消息服务 // TODO后续可考虑注入点的方式，统一spring message
 * @author songxy DateTime 2017年5月27日 上午11:48:22
 * @company winter
 * @email thinkdata@163.com
 * @version 1.0
 */
@SuppressWarnings({ "unchecked", "rawtypes" })
public class KafkaService implements InitializingBean, DisposableBean {

	private final static Logger LOGGER = LoggerFactory
			.getLogger(KafkaService.class);
	private static final int kAFKA_SEND_TIMEOUT = 2000;

	private KafkaProducer producer;
	private KafkaProducerFactory producerFactory;

	public KafkaService(KafkaProducerFactory producerFactory) {
		if (producerFactory == null) {
			throw new IllegalArgumentException("KafkaProducerFactory is null.");
		}
		this.producerFactory = producerFactory;
		this.producerFactory.start();
		this.producer = this.producerFactory.getProducer();
	}

	public <K, V> void send(String topic, K key, V value) {
		sendMessage(topic, null, key, value);
	}

	public <K, V> void sendMessage(String topic, Integer partition, K key,
			V value) {
		try {
			producer.send(new ProducerRecord<K, V>(topic, partition, key, value));
		} catch (Exception e) {
			LOGGER.error(e.getMessage(), e);
		} finally {
		}
	}

	/**
	 * 
	 * @description: 创建主题下的消息
	 * @author songxy DateTime 2017年5月27日 上午11:48:04
	 * @param topic
	 * @param key
	 * @param value
	 * @return
	 */
	public <K, V> ProducerRecord<K, V> createProducerRecord(String topic,
			K key, V value) {
		return new ProducerRecord<>(topic, key, value);
	}

	/**
	 * 
	 * @description: 发送消息
	 * @author songxy DateTime 2017年5月27日 上午11:47:48
	 * @param record
	 * @return
	 */
	public <K, V> Result<K, V> sendMessage(ProducerRecord record) {
		try {
			final ListenableFuture<Result<K, V>> future = new SettableListenableFuture();
			producer.send(record, createCallback(record, future));
			return future.get(kAFKA_SEND_TIMEOUT, TimeUnit.MILLISECONDS);
		} catch (InterruptedException | ExecutionException | TimeoutException e) {
			LOGGER.error(e.getMessage(), e);
			return null;
		} finally {
		}
	}

	private ProducerCallback createCallback(ProducerRecord record,
			ListenableFuture listenableFuture) {
		return new ProducerCallback(record, listenableFuture);
	}

	private interface Listener<K, V> {
		void onSuccess(String topic, Integer partition, K key, V value,
				RecordMetadata recordMetadata);

		void onError(String topic, Integer partition, K key, V value,
				Exception exception);

	}

	@SuppressWarnings("rawtypes")
	private class ProducerCallback implements Callback {

		private ListenableFuture result;
		private ProducerRecord record;

		public ProducerCallback(ProducerRecord record, ListenableFuture result) {
			this.record = record;
			this.result = result;
		}

		@Override
		public void onCompletion(RecordMetadata metadata, Exception ex) {
			if (ex == null) {
				((SettableListenableFuture) result).set(new Result(record,
						metadata));
				if (LOGGER.isDebugEnabled())
					LOGGER.debug("send compeleted {} .", record);
			} else {
				((SettableListenableFuture) result).setException(ex);
				if (LOGGER.isDebugEnabled())
					LOGGER.error("send error message {}. because of {}.",
							metadata != null ? metadata.toString() : "",
							ex.getMessage());
			}
		}

	}

	private class Result<K, V> {

		private final ProducerRecord<K, V> producerRecord;

		private final RecordMetadata recordMetadata;

		public Result(ProducerRecord<K, V> producerRecord,
				RecordMetadata recordMetadata) {
			this.producerRecord = producerRecord;
			this.recordMetadata = recordMetadata;
		}

		public ProducerRecord<K, V> getProducerRecord() {
			return this.producerRecord;
		}

		public RecordMetadata getRecordMetadata() {
			return this.recordMetadata;
		}
	}

	public interface IProducerFactory<K, V> {

		KafkaProducer<K, V> createKafkaProducer();

	}

	@Override
	public void destroy() throws Exception {
		this.producerFactory.stop();
		this.producer.close();
	}

	@Override
	public void afterPropertiesSet() throws Exception {
	}
}
